/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.parser.mp.mpost;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.RESessionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSessionBuilder;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.blockingQueueSize;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.coreThreadPoolSize;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.intervaltime;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.jsonThreadSize;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.maxThreadPoolSize;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.mpostQueue;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.mpost_topic;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.nameserver;
import static cn.ac.iie.di.parser.mp.main.ParserMpMain.timeout;
import cn.ac.iie.di.parser.mp.utils.LoadTableMapData;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class MPostHandler implements Runnable {

    static private final Logger logger = Logger.getLogger(MPostHandler.class.getName());

    ThreadPoolExecutor threadpool = null;
    private String tbname = "mpost";

    public MPostHandler() {
        //创建一个线程池
        threadpool = new ThreadPoolExecutor(coreThreadPoolSize, maxThreadPoolSize, 3, TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>(blockingQueueSize),
                new ThreadFactory() {
                    AtomicInteger idx = new AtomicInteger(0);

                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "MPostThreadPoolTask_" + idx.getAndIncrement());
                    }
                }, new RejectedExecutionHandler() {

                    @Override
                    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                        while (true) {
                            try {
                                executor.getQueue().put(r);
                                break;
                            } catch (InterruptedException ex) {
                                logger.warn("", ex);
                            }
                        }
                    }
                });
    }

    @Override
    public void run() {
        RESendSession session = null;
        SessionPool pool;

        try {
            //初始化RESendSessionBuild
            logger.info("the load table is " + tbname + " and the mqnameserver is " + nameserver + " and the mqtopic is " + mpost_topic);
            REConnection conn = new REConnection(nameserver);
            RESendSessionBuilder builder = (RESendSessionBuilder) conn.getSendSessionBuilder(mpost_topic);
            //mpost.mapdata文件放置消息队列的key和类型
            HashMap columnsMap = LoadTableMapData.dataMapHashMaps.get(tbname);
            for (Iterator columnsit = columnsMap.entrySet().iterator(); columnsit.hasNext();) {
                Map.Entry<String, String> entry = (Map.Entry<String, String>) columnsit.next();
                logger.info("column name = " + entry.getKey() + ", type = " + entry.getValue());
                if (entry.getValue().trim().equalsIgnoreCase("string")) {
                    builder.addColumn(entry.getKey(), REFieldType.String, true);
                } else if (entry.getValue().trim().equalsIgnoreCase("long")) {
                    builder.addColumn(entry.getKey(), REFieldType.Long, true);
                } else if (entry.getValue().trim().equalsIgnoreCase("strings")) {
                    builder.addColumn(entry.getKey(), REFieldType.Strings, true);
                } else if (entry.getValue().trim().equalsIgnoreCase("int")) {
                    builder.addColumn(entry.getKey(), REFieldType.Int, true);
                } else if (entry.getValue().trim().equalsIgnoreCase("struct")) {
                    builder.addColumn(entry.getKey(), REFieldType.Struct, true);
                    HashMap<String, REFieldType> detailMap = new HashMap<>();
                    detailMap.put("tct_l1", REFieldType.Long);
                    detailMap.put("tct_l2", REFieldType.Long);
                    detailMap.put("tct_l3", REFieldType.Long);
                    REFieldType.Struct.setDetail(entry.getKey(), detailMap);
                } else if (entry.getValue().trim().equalsIgnoreCase("structs")) {
                    builder.addColumn(entry.getKey(), REFieldType.Structs, true);
                    HashMap<String, REFieldType> detailMap = new HashMap<>();
                    detailMap.put("v", REFieldType.String);
                    detailMap.put("w", REFieldType.String);
                    REFieldType.Structs.setDetail(entry.getKey(), detailMap);
                } else if (entry.getValue().trim().equalsIgnoreCase("int[]")) {
                    builder.addColumn(entry.getKey(), REFieldType.Ints, true);
                } else {
                    logger.warn(entry.getKey() + ": (type)" + entry.getValue() + " do not recognize this type!");
                }
            }
            pool = new SessionPool(builder);
            session = (RESendSession) builder.build();
            session.start();

        } catch (Exception e) {
            logger.error("session.build error,cause " + e.getMessage(), e);
            return;
        }
        logger.info("init RESendSession finish.");
        //循环从mpostQueue获取josn，启动MPostThreadPoolTask线程进行发送
        while (true) {
            try {
                int addCount = 0;
                ArrayList<String> jsonList = new ArrayList<>(jsonThreadSize);
                for (int i = 0; i < jsonThreadSize; ++i) {
                    String str = mpostQueue.poll(intervaltime, TimeUnit.MILLISECONDS);
                    if (str == null) {
                        break;
                    }
                    jsonList.add(str);
                    addCount++;
                }
                if (jsonList.isEmpty()) {
                    logger.info("ArrayBlockingQueue doesn't hava available data, and sleep 5 seconds.");
                    Thread.sleep(5000);
                } else {
                    logger.info("get " + addCount + " JSONData from ArrayBlockingQueue.");
                    threadpool.execute(new MPostThreadPoolTask(jsonList, pool, timeout));
                }
            } catch (Exception ex) {
                logger.error("get JSONData from ArrayBlockingQueue failed,cause :" + ex.getMessage(), ex);
            }
        }

    }

    static public class SessionPool {

        RESendSessionBuilder builder;
        LinkedList<RESendSession> pool;

        public SessionPool(RESendSessionBuilder builder) {
            this.builder = builder;
            pool = new LinkedList<>();
        }

        public synchronized RESendSession borrowSession() throws RESessionException, REConnectionException {
            try {
                return pool.removeFirst();
            } catch (NoSuchElementException ex) {
                RESendSession rtn = (RESendSession) builder.build();
                rtn.start();
                return rtn;
            }
        }

        public synchronized void returnSession(RESendSession session) {
            pool.add(session);
        }
    }

}
